package com.atguigu.flink.state;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

/**
 * Created by Smexy on 2023/3/1
 *
 *  barrier是否对齐，只在多流操作的场景下有意义，单流无需考虑。
 *
 *  barrier对齐：  多个流的同一批的barrier都已经到达，才对state进行持久化。
 *         当第一个barrier到达后，是否阻塞barrier已经到达的流中的数据。
 *              阻塞：   EOS.
 *                          会存在反压，最终导致 数据会有积压，实时性打折扣。
 *
 *              不阻塞:  AT LEAST ONCE
 *
 *         致命的问题： barrier向下游传递的时效会被拖长，导致checkpoint可能会超时。
 *                      如果连续的多个批次的ck超时，会导致之前备份的进度，远远落后于数据处理的进度，恢复起来，时间长。
 *     -----------------------------------
 * barrier不对齐：  当某一批的barrier到达算子的输入缓存时，就立刻触发checkpoint，把barrier直接发送到下游。
 *                  将两个流 输出缓存区和barrier之间的所有缓存的数据及状态都进行备份。
 *
 *                  失败时，按照备份，将 将两个流 输出缓存区和barrier之间的所有缓存的数据及状态 恢复原样。
 *
 *              优势：   增快了barrier的流动，减少ck的超时概率。
 *
 *              劣势:    增加了IO
 *
 *              是EOS
 *
 *   ----------------------
 *      默认： 采用barrier对齐，EOS
 *
 *      设置为 barrier对齐，at least once ：
 *          env.enableCheckpointing(2000, CheckpointingMode.AT_LEAST_ONCE);
 *
 *
 *       Unaligned checkpoints can only be used with checkpointing mode EXACTLY_ONCE
 *
 *       maxConcurrentCheckpoints can't be > 1 if UnalignedCheckpoints enabled

 */
public class Demo11_CheckPoint
{

    public static void main(String[] args) {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 3333);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        //默认 EOS 会有反压
        //env.enableCheckpointing(2000);
        // at least once
        env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);

        CheckpointConfig checkpointConfig = env.getCheckpointConfig();
        checkpointConfig.setCheckpointStorage("file:///d:/tmp");
        //开启 barrier 不对齐
        checkpointConfig.enableUnalignedCheckpoints();

        //设置ck的超时时间，从checkpointCoordinator触发本次ck请求开始，查过了这个时间，这个批次的ck还未完成，就作废
        checkpointConfig.setCheckpointTimeout(5l*60000);

        //两次ck的最小间隔
        checkpointConfig.setMinPauseBetweenCheckpoints(5000);

        //流中同时最多存在多少个barrier  非对齐，这个值只能是1
        checkpointConfig.setMaxConcurrentCheckpoints(1);

        //默认job 被cacel，清理之前ck的数据
        checkpointConfig.setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

        env
            .socketTextStream("hadoop103", 8888)
            .map(new MyMapFunction())
            .addSink(new SinkFunction<String>()
            {
                //如果数据中包含了 x，手动模拟异常
                @Override
                public void invoke(String value, Context context) throws Exception {

                    if (value.contains("x")){
                        throw new RuntimeException("出错了...");
                    }
                    System.out.println(value);
                }
            });


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    //累积输入的每个字符串
    public static class MyMapFunction implements MapFunction<String,String>, CheckpointedFunction
    {

        private ListState<String> listState;

        @Override
        public String map(String value) throws Exception {
            listState.add(value);
            return listState.get().toString();
        }

        @Override
        public void snapshotState(FunctionSnapshotContext context) throws Exception {
            System.out.println("MyMapFunction.snapshotState");

        }

        //初始化状态。  在Task重启时，从之前的备份中帮你恢复状态。 只在Task创建时，执行一次！
        @Override
        public void initializeState(FunctionInitializationContext context) throws Exception {
            System.out.println("MyMapFunction.initializeState");

            listState = context.getOperatorStateStore().getListState(new ListStateDescriptor<>("strs", String.class));
        }
    }

}
